package se.kth.ict.id2203.uniformconsensus;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import se.kth.ict.id2203.abortableconsensus.AbortableConsensus;
import se.kth.ict.id2203.abortableconsensus.AcPropose;
import se.kth.ict.id2203.abortableconsensus.AcReturn;
import se.kth.ict.id2203.beb.BebBroadcast;
import se.kth.ict.id2203.beb.BestEffortBroadcast;
import se.kth.ict.id2203.eld.EventualLeaderDetector;
import se.kth.ict.id2203.eld.Trust;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.address.Address;
import se.sics.kompics.launch.Topology;


public class PaxosUniformConsensus extends ComponentDefinition {
	
	Negative<UniformConsensus> ucPort = provides(UniformConsensus.class);
	Positive<BestEffortBroadcast> beb = requires(BestEffortBroadcast.class);
	Positive<AbortableConsensus> ac = requires(AbortableConsensus.class);
	Positive<EventualLeaderDetector> eld = requires(EventualLeaderDetector.class);
	
	private Address self;
	private Topology topology;
	private Set<Integer> seenIds;
	private boolean isLeader;
	
	private Map<Integer,Integer> proposal;
	private Map<Integer,Boolean> proposed;
	private Set<Integer> decided;
	
	/*
	 * it would be enough with a set since it is true/false but I'll
	 * keep this as a map just to be able to duplicate the algorithm closer.
	 */
	
	private static final Logger logger =
			LoggerFactory.getLogger(PaxosUniformConsensus.class);
	
	public PaxosUniformConsensus() {
		logger.info("PaxosUniformConsensus created.");
		subscribe(handleInit, control);
		subscribe(handleTrust, eld);
		subscribe(handleUcPropose, ucPort);
		subscribe(handleAcReturn,ac);
		subscribe(handleDecided, beb);
	}
	
	Handler<PaxosUniformConsensusInit> handleInit = new Handler<PaxosUniformConsensusInit>() {
		public void handle(PaxosUniformConsensusInit event) {
			
			logger.info("PaxosUniformConsensus started.");
			topology = event.getTopology();
			self = topology.getSelfAddress();
			
			seenIds = new HashSet<Integer>();
			isLeader = false;
			
			proposal = new HashMap<Integer,Integer>();
			proposed = new HashMap<Integer,Boolean>();
			decided = new HashSet<Integer>();
		}
	};
	
	/*
	 * procedure initInstance(id)
	 */
	public void initInstance( int id )
	{
		logger.info("initInstance");
		// if (id /∈ seenIds) then
		if ( !seenIds.contains( id ) )
		{
			// proposal[id] := ⊥;
			// proposed[id] := decided[id] := false;
			proposed.put(id, false);
			// seenIds := seenIds ∪ {id};
			seenIds.add(id);
		}
	}
		
	/*
	 * upon event ⟨ trust | pi ⟩ do
	 */
	Handler<Trust> handleTrust = new Handler<Trust>() {
		public void handle(Trust event) 
		{
			logger.info("handleTrust");
			// if (pi = self) then
			if ( event.getTrustedSource().getId() == self.getId() )
			{
				isLeader = true;
				// for all id ∈ seenIds do
				// tryPropose(id);
				for ( Integer id : seenIds )
				{
					tryPropose(id);
				}
			}
			else
			{
				isLeader = false;
			}
		}
	};
	

	/* 
	 * upon event ⟨ ucP ropose | id, v ⟩ do
	 */
	Handler<UcPropose> handleUcPropose = new Handler<UcPropose>() {
		public void handle(UcPropose event) 
		{
			logger.info("handleUcPropose");
			int id = event.getId();
			logger.info("Proposal received: id="+id +", value="+event.getProposal());
			
			// initInstance(id);
			initInstance(id);
			// proposal[id] := v;
			proposal.put(id, event.getProposal());
			// tryPropose(id);
			tryPropose(id);
		}

	};
	
	/*
	 * procedure tryPropose(id)
	 */
	public void tryPropose( int id )
	{
		logger.info("tryPropose started" );
		// if (leader = true ∧ proposed[id] = false ∧ proposal[id] ̸= ⊥) then
		if ( isLeader && (proposed.get(id) == false ) && proposal.containsKey(id) )
		{
			// proposed[id] := true;
			proposed.put(id, true);
			// trigger ⟨ acPropose | id, proposal[id] ⟩;
			trigger(new AcPropose(id,proposal.get(id)), ac);
		}
		
	}
	
	
	/* 
	 * upon event ⟨ acReturn | id, result ⟩ do
	 */
	Handler<AcReturn> handleAcReturn = new Handler<AcReturn>() {
		public void handle(AcReturn event) 
		{
			logger.info("handleAcReturn");
			int id = event.getId();
			//if (result ̸= ⊥) then
			if ( event.getResult() != null )
			{
				//trigger ⟨ bebBroadcast | [Decided, id, result] ⟩;
//				trigger(new Decided(self,id,event.getResult()), beb);
				trigger(new BebBroadcast(new Decided(self,id,event.getResult())), beb);
				
			}			
			//else
			else
			{
				//proposed[id]:= false;
				proposed.put(id, false);
				//tryPropose(id);
				tryPropose(id);
			}

		}

	};
	
	
	/* 
	 * upon event ⟨ bebDeliver | pi , [Decided, id, v] ⟩ do
	 */
	Handler<Decided> handleDecided = new Handler<Decided>() {
		public void handle(Decided event) 
		{
			logger.info("handleDecided");
			int id = event.getId();
			// initInstance(id);
			initInstance(id);
			// if (decided[id] = false) then
			if ( !decided.contains(id) )
			{
				// decided[id] := true;
				decided.add(id);
				// trigger ⟨ ucDecide | id, v ⟩;
				trigger(new UcDecide(id, event.getDecision()), ucPort);
			}

		}
		
	};

	
	
}
